package org.example
import org.apache.spark.sql.SparkSession

import java.util.regex.Pattern
object class0320 {
  def main(args: Array[String]): Unit = {
    val spark =SparkSession
      .builer()
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc=spark.sparkContext

    val filePath = "src/main/resources/"
    val moviesRDD = sc.textFile(filePath + "movaies.dat")
    val occupationsRDD = sc.textFile(filePath + "occupations.dat")
    val ratingsRDD = sc.textFile(filePath + "ratings.dat")
    val usersRDD = sc.textFile(filePath + "users.dat")
    println("总电影数："+moviesRDD.count())
    println("总用户数："+usersRDD.count())
    println("评分条数："+ratingsRDD.count())
    println("职业数："+occupationsRDD.count())


    val userRating = ratingsRDD.map(_.split("::"))
      .map(user => (user(1),user(0)))
      .filter(_._2.equals("2688"))
    println("用户ID为2688的影评数量为：" + userRaing.count() + "以下为电影详情\n")

    val moveInfo =moviesRDD.map(_.split("::"))
      .map(movie => (movie(0),(movie(1)),(movie(2))))
      .join(userRating)
      .map(item => (item._1,item._2._2,item._2._1._1,item._2_1._2))
    movieInfo.foreach(println)


    //zuoye
    val rating = ratingsRDD.map(x => x.split("::")).map {
      x => {   (x(0), x(1), x(2))
      }
    }.cache()
    println("平均得分最高的前 10 名的电影名称简单版")
    rating.map(x => (x._2, (x._3.toDouble, 1)))
      .reduceByKey((x, y) => {
       (x._1 + y._1, x._2 + y._2)
      })
      .map(x => (x._2._1 / x._2._2, x._1))
      .sortByKey(false)
      .take(5)
      .foreach(println)
    println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次 数))")
    val moviesInfo = moviesRDD.map(x => x.split("::"))
      .map(x => {
        (x(0), (x(1), x(2)))
      })
    val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
    .reduceByKey((x, y) => {
      (x._1 + y._1, x._2 + y._2)
    })
      .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2)))
    moviesInfo.join(ratingsInfo)
      .map(info => {
        (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2,info._2._2._3))
      }).sortByKey(false)
      .take(5)
      .foreach(println)
    println("观影人数最多的前 10 部电影")
    val watchViewsInfo = rating.map(x => {
      (x._2, 1)
    }).reduceByKey((x, y) => x + y)
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(5)
    watchViewsInfo.foreach(println(_))
    println("===================>")
    rating.map(x => (x._2, 1))
      .reduceByKey((x, y) => {
        (x + y)
    })
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(10)
      .foreach(println)
    println("详情的输出(  观影人数，电影编号)")
    moviesInfo.join(ratingsInfo).map(x => {
      (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
    }).sortByKey(false)
      .take(10)
      .foreach(println)

    println(movieInfo)

    //0411
    val movieStyle = moviesRDD.map(_.split("::"))
      .map(x => x(0),x(2))
      .flatMapValues(x => x.split("\\|"))
      .map(x => (x._2,1))
    //在控制台打印科幻片的电影数量（Fantasy）
      .reduceByKey((x,y) => x + y)
    movieStyle.take(10).foreach(println)
    //正则表达式
    val pattern = Pattern.compile("(.*)(\\(\\d{4}\\))")

    val movieYear = moviesRDD.map(_.split("::"))
      .map(x => (x(1),1))
      .map(x =>{
        var name = ""
        var year = ""
        val matcher = pattern.matcher(x._1)
        if (matcher.find()) {
        name = matcher.group(1)
        year = matcher.group(2)
        year = year.substring(1,year.length -1)
        }
        if (year == ""){
          (-1,1) //标识码
        }else{
          (year.toInt,1)
        }
      })
      .reduceByKey((x,y) => x+y)


        //求2000年度电影生产数量
        .sortByKey()
        .collect()
    movieYear.foreach(println)



    sc.stop()
  }

}
